package studio.raptor.extractor.segment.task;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

import org.apache.http.client.HttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.EventDataInfo;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.task.assemble.EventMetaMsgTask;

/**
 * Created by liuyong8 on 2017/7/24.
 */
@Component
@Scope("prototype")
public class NodeMsgAssemblingTask implements Runnable {
  private static final Logger log = LoggerFactory.getLogger(NodeMsgAssemblingTask.class);

  @Autowired
  private HttpClient httpClient;

  private ServerConfig serverConfig;

  private BlockingQueue<EventMeta> eventMetaQueue;

  private BlockingQueue<EventDataInfo> eventDataQueue;

  private ExecutorService exec;

  private int count;

  @Override
  public void run() {
    while (!Thread.currentThread().isInterrupted()) {
      try {
        CompletionService<EventDataInfo> eventMetaComp =
            new ExecutorCompletionService<EventDataInfo>(exec);
        int concurrentExtractEvents = 10;
        List<EventMeta> eventMetas = new ArrayList<EventMeta>();
        List<EventDataInfo> eventDatas = new ArrayList<EventDataInfo>();

        // 这一步从队列中拿出来的数量要尽量少，暂定10个
        int nums = eventMetaQueue.drainTo(eventMetas, concurrentExtractEvents);
        if (nums <= 0) {
          printPerfTip();
          LockSupport.parkUntil(System.currentTimeMillis() + 100);
        }else {
          for (EventMeta eventMeta : eventMetas) {
            eventMetaComp
                    .submit(new EventMetaMsgTask(serverConfig.getServerUrlMap(), eventMeta, exec, httpClient));
          }

          for (int i = 0; i < nums; i++) {
            Future<EventDataInfo> future = eventMetaComp.poll(serverConfig.getOverTime(), TimeUnit.MILLISECONDS);
            if (future == null) {
              break;
            } else {
              try {
                EventDataInfo eventDataInfo = future.get();
                eventDatas.add(eventDataInfo);
                eventMetas.remove(eventDataInfo.getEventMeta());
              } catch (ExecutionException e) {
                // 异常算超时获取
                continue;
              }
            }
          }

          for (EventMeta eventMeta : eventMetas) {
            EventDataInfo eventDataInfo = new EventDataInfo();
            eventDataInfo.setMetaId(eventMeta.getMetaId());
            eventDataInfo.setEventMeta(eventMeta);
            eventDatas.add(eventDataInfo);
          }

          // 排序
          Collections.sort(eventDatas);

          for (EventDataInfo eventDataInfo : eventDatas) {
            eventDataQueue.put(eventDataInfo);
          }
        }

      } catch (InterruptedException e) {

      }
    }
  }

  private void printPerfTip(){
    if ((++count) % 1000 == 0) {
      log.warn("extracted events have been assembled, please speed up the extracting");
      count = 0;
    }
  }

  public ServerConfig getServerConfig() {
    return serverConfig;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }

  public BlockingQueue<EventMeta> getEventMetaQueue() {
    return eventMetaQueue;
  }

  public void setEventMetaQueue(BlockingQueue<EventMeta> eventMetaQueue) {
    this.eventMetaQueue = eventMetaQueue;
  }

    public BlockingQueue<EventDataInfo> getEventDataQueue() {
        return eventDataQueue;
    }

    public void setEventDataQueue(BlockingQueue<EventDataInfo> eventDataQueue) {
        this.eventDataQueue = eventDataQueue;
    }

    public ExecutorService getExec() {
        return exec;
    }

    public void setExec(ExecutorService exec) {
        this.exec = exec;
    }
}
